Windows Azure: Building a Secure Backup System (part 6) - Uploading Efficiently Using Blocks

def put_block(self, container_name, blob_name, block_id, data):

        # Take a block id and construct a URL-safe, base64 version
        base64_blockid = base64.encodestring(str(block_id)).strip()
        urlencoded_base64_blockid = urllib.quote(base64_blockid)

        # Make a PUT request with the block data to blob URI followed by
        # ?comp=block&blockid=<blockid>
        return self._do_store_request("/" + container_name + "/" + \
                                  blob_name + \
                                          "?comp=block&blockid=" + \
                                          urlencoded_base64_blockid, \
                                          'PUT', {}, data)

    def put_block_list(self, container_name, blob_name, \
                       block_list, content_type):
        headers = {}
        if content_type is not None:
            headers["Content-Type"] = content_type

        # Begin XML content
        xml_request = "<?xml version=\"1.0\" encoding=\"utf-8\"?><BlockList>"

        # Concatenate block ids into block list
        for block_id in block_list:
            xml_request += "<Block>" + \
            base64.encodestring(str(block_id)).strip() + "</Block>"

        xml_request += "</BlockList>"

        # Make a PUT request to blob URI followed by ?comp=blocklist
        return self._do_store_request("/" + container_name + \
                                "/" + blob_name + \
                                          "?comp=blocklist", 'PUT',\
                                 headers, xml_request)

					  

def upload_archive(data, filename, account, key):
    conn = storage.Storage("blob.core.windows.net",account, key)

    # Try and create container. Will harmlessly fail if already exists
    conn.create_container("enc", False)

    # Heuristics for blocks
    # We're pretty hardcoded at the moment. We don't bother using blocks
    # for files less than 4MB.
    if len(data) < 0:# 4 * 1024 * 1024:
        resp = conn.put_blob("enc", filename, data,"application/octet-stream")
    else:
        resp = upload_archive_using_blocks(data, filename, conn)

    if not (resp.status >=200 and resp.status < 400):
        # Error! No error handling at the moment
        print resp.status, resp.reason, resp.read()
        sys.exit(1)


def upload_archive_using_blocks(data, filename, conn):

    blocklist=[]

    queue = Queue.Queue()
    if parallel_upload:
        # parallel_upload specifies whether blocks should be uploaded
        # in parallel and is set from the command line.
        for i in range(num_threads):
            t = task.ThreadTask(queue)
            t.setDaemon(True) # Run even without workitems
            t.start()

    offset =0

    # Block uploader function used in thread queue
    def block_uploader(connection, block_id_to_upload,\
                     block_data_to_upload):
        resp = connection.put_block("enc", filename, block_id_to_upload,\
                                          block_data_to_upload)
        if not( resp.status>=200 and resp.status <400):
            print resp.status, resp.reason, resp.read()
            sys.exit(1) # Need retry logic on error

    while True:

        if offset>= len(data):
            break

        # Get size of next block. Process in 4MB chunks
        data_to_process = min( 4*1024*1024, len(data)-offset)

        # Slice off next block. Generate an SHA-256 block id
        # In the future, we could use it to see whether a block
        # already exists to avoid re-uploading it

        block_data = data[offset: offset+data_to_process]
        block_id =  hashlib.sha256(block_data).hexdigest()
        blocklist.append(block_id)

        if parallel_upload:
           # Add work item to the queue.
            queue.put([block_uploader, [conn, block_id, block_data]])
        else:

            block_uploader(conn, block_id, block_data)

        # Move i forward
        offset+= data_to_process

    # Wait for all block uploads to finish
    queue.join()

    # Now upload block list
    resp = conn.put_block_list("enc", filename, \
            blocklist, "application/octet-stream")
    return resp